#include <sys/msg.h>
#include <pthread.h>
#include <iostream>
#include "EventGateway.h"
#include "FileTrans.h"
#include "../QCommon/csmsq.h"
#include <sys/stat.h>
#include <iostream>

using namespace std;

void* send(void *arg)
{
    EventOut *eventOut=(EventOut*)arg;
	int msqid=retrieve_msqid();
	while(1)
	{
	    char *event=new char[MAX_EVENT_SIZE];
		msgrcv(msqid,event,MAX_EVENT_SIZE,0,0);
		AbstractEvent *e=(AbstractEvent*)event;
        if(e->type==file_ask)
        {
            FileAskEvent *fae=(FileAskEvent*)e;
            FileOut::getInstance().send(*fae);
        }
        eventOut->trans.send_msg(*e);
	}
	return NULL;
}

void EventOut::start_send()
{
    pthread_t tid;
    pthread_create(&tid,NULL,send,this);
}

void EventOut::send_msg(AbstractEvent &event)
{
    events.push(event);
}

EventOut& EventOut::getInstance()
{
    static EventOut eventOut;
    return eventOut;
}

EventIn& EventIn::getInstance()
{
    static EventIn eventIn;
    return eventIn;
}

static CSmsq msq;

struct RecArg
{
    CSmsq *msqs;
    EventIn *in;
};

void* recv(void *arg)
{
    RecArg *rArg=(RecArg*)arg;
    EventIn *eventIn=rArg->in;
    CSmsq *msqs=rArg->msqs;
	int msqid;//Get from hash
	while(1)
	{
		AbstractEvent *event=eventIn->trans.recv_msg();
		if(event!=NULL)
		{
		    if(event->type==file_answer)
		    {
		        FileAnswerEvent *fae=(FileAnswerEvent*)event;
		        FileOut::getInstance().ack(*fae);
		    }
            else if(event->type==network)
		    {
                int *pids = msqs->getpids();
                for(int i=0;i<msqs->size();i++)
                {
                    msqid=msqs->get(pids[i]);
                    if(msqid==0)
                        continue;
                    msgsnd(msqid,event,size_event(*event)-sizeof(long),0);

                }
		    }
		    else
		    {
		        msqid=msqs->get(event->dest_pid);
                if(msqid==0)
                    continue;
                msgsnd(msqid,event,size_event(*event)-sizeof(long),0);
		    }
		}
	}
	return NULL;
}


void* check_client(void *arg)
{
    RecArg *rArg=(RecArg*)arg;
    CSmsq *msqs=rArg->msqs;

    int fd=open(Q_CS_FIFO,O_RDWR);
    struct CSData data;
    while(1)
    {
        read(fd,&data,sizeof(CSData));
        msqs->put(data.pid,data.msqid);
        //cout<<"pid:"<<data.pid<<"msqid;"<<data.msqid<<endl;
        ACK ack;
        ack.type=1;
        ack.res='y';
        msgsnd(data.msqid,&ack,sizeof(char),0);
        sleep(1);
    }
    delete rArg;
    close(fd);
    return NULL;
}

void EventIn::start_recv()
{
    pthread_t tid;
    RecArg *arg=new RecArg;
    arg->in=this;
    arg->msqs=&msq;
    pthread_create(&tid,NULL,recv,arg);
    pthread_create(&tid,NULL,check_client,arg);
}
